package com.swak.paxos.impl;

import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.paxos.Group;
import com.swak.paxos.PaxosConfig;
import com.swak.paxos.PaxosConfig.GroupConfig;
import com.swak.paxos.Value;
import com.swak.paxos.exception.PaxosException;
import com.swak.paxos.protocol.PaxosForward;
import com.swak.paxos.statemachine.master.MasterStateMachine;
import com.swak.paxos.statemachine.system.SystemxStateMachine;
import com.swak.paxos.store.PaxosStorage;
import com.swak.reactivex.cluster.Member;
import com.swak.reactivex.threads.EventLoop;
import com.swak.reactivex.threads.SingleThreadEventLoop;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.reactivex.threads.waitstrategy.WakeupWaitStrategy;
import com.swak.reactivex.transport.Message;

import lombok.Getter;

/**
 * 分组： 每个分组可以代表一个业务
 * 
 * @author DELL
 */
@Getter
public class PaxosGroup implements Group {

	private static final Logger logger = LoggerFactory.getLogger(Group.class);
	private static final CompletableFuture<Void> IMFollowerErrorFuture = CompletableFuture
			.failedFuture(new PaxosException("Im Follower, Can Not Commit Value;"));
	private static final CompletableFuture<Void> IMNotInMemberShipErrorFuture = CompletableFuture
			.failedFuture(new PaxosException("Im Not In MemberShip, can not commit value;"));

	private final int groupID;
	private final GroupConfig config;
	private final PaxosGroupUnsafe unsafe;
	private final PaxosNode node;
	private final EventLoop eventLoop;
	private final PaxosStorage storage;
	private final PaxosInstance instance;

	/**
	 * 系统状态
	 */
	private final SystemxStateMachine systemSM;
	private final MasterStateMachine masterSM;

	public PaxosGroup(int groupID, PaxosConfig config, GroupConfig groupConfig, PaxosNode node, PaxosStorage storage) {
		this.groupID = groupID;
		this.config = groupConfig;
		this.node = node;
		this.storage = storage;
		this.systemSM = new SystemxStateMachine(node.id(), groupConfig);
		this.masterSM = new MasterStateMachine(groupID, config, node);
		this.eventLoop = new SingleThreadEventLoop(new SwakThreadFactory("Paxos.Group-" + groupID, false),
				new WakeupWaitStrategy());
		this.instance = new PaxosInstance(config, this.node, this, this.storage, this.eventLoop, systemSM, masterSM);
		this.unsafe = new PaxosGroupUnsafe(this.eventLoop, this.instance);
	}

	@Override
	public int getGroupID() {
		return this.groupID;
	}

	@Override
	public boolean isIMFollower() {
		return this.config.isIMFollower();
	}

	@Override
	public boolean isIMMaster() {
		return this.masterSM.isIMMaster();
	}

	@Override
	public Set<String> getFollowers() {
		return this.systemSM.getFollowerShips();
	}

	@Override
	public CompletableFuture<Void> propose(Value value) {
		return this.propose(value, false);
	}

	@Override
	public CompletableFuture<Void> propose(Value value, boolean useMaster) {

		/*
		 * Follower 节点不能提交任务
		 */
		if (this.isIMFollower()) {
			if (logger.isDebugEnabled()) {
				logger.debug("I am Follower, can not propose value.");
			}
			return IMFollowerErrorFuture;
		}

		/*
		 * 不是有效的集群节点，不能提交数据
		 */
		if (!this.systemSM.isIMInMembership()) {
			if (logger.isDebugEnabled()) {
				logger.debug("I am not in MemberShip, can not propose value.");
			}
			return IMNotInMemberShipErrorFuture;
		}

		/*
		 * 不需要使用 Master, 或者自己是 Master, 没有 Master, Master 节点不是同步集合
		 */
		if (!useMaster || (useMaster && this.isIMMaster()) || this.masterSM.noMaster()
				|| this.systemSM.inMemberShip(this.masterSM.getMasterNodeID())) {
			return this.instance.propose(value);
		}

		/*
		 * 转发此消息到 Master 节点： 如果节点不存在会抛出异常
		 */
		this.node.sendMessage(this.masterSM.getMasterNodeID(),
				new PaxosForward().setGroupId(this.groupID).setValue(value));

		/*
		 * 提交一个: 伪消息,此消息不会真实的提交，但是会等待提交成功， 或者超时
		 */
		return this.instance.propose(value.clone().addFlag(Value.Flag_NOSEND));
	}

	@Override
	public void onMemberShipChanaged(Member member, GroupConfig config, List<Member> members, boolean add) {
		if (logger.isDebugEnabled()) {
			logger.debug("{} 节点, NodeID:{}", add ? "添加" : "删除", member.id().getId());
		}
		this.unsafe.onMemberShipChanaged(member, config, members, add);
	}

	@Override
	public void onReceiveMessage(Message message) {
		this.unsafe.onReceiveMessage(message);
	}

	@Override
	public void close() {
		try {
			this.eventLoop.shutdownGracefully();
		} catch (Exception e) {
		}
		try {
			this.storage.close();
		} catch (Exception e) {
		}
	}

	/**
	 * 在 EventLoop 下执行如下的业务
	 * 
	 * @author 618lf
	 */
	class PaxosGroupUnsafe implements Unsafe {

		private final EventLoop eventLoop;
		private final PaxosInstance instance;

		PaxosGroupUnsafe(EventLoop eventLoop, PaxosInstance instance) {
			this.eventLoop = eventLoop;
			this.instance = instance;
		}

		@Override
		public void onMemberShipChanaged(Member member, GroupConfig config, List<Member> members, boolean add) {
			if (this.eventLoop.inEventLoop()) {
				this.doOnMemberShipChanaged(member, config, members, add);
			} else {
				this.eventLoop.execute(() -> {
					this.onMemberShipChanaged(member, config, members, add);
				});
			}
		}

		private void doOnMemberShipChanaged(Member member, GroupConfig config, List<Member> members, boolean add) {

			/*
			 * 添加节点
			 */
			if (add) {

				// 添加节点信息
				PaxosGroup.this.systemSM.addMember(member.id().getId(), config);

				/*
				 * 启动一半的节点时
				 */
				if (PaxosGroup.this.systemSM.isValidMembership()) {
					PaxosGroup.this.masterSM.checkBeMaster();
				}
			}

			/*
			 * 删除节点
			 */
			else {
				PaxosGroup.this.systemSM.removeMember(member.id().getId(), config);
				PaxosGroup.this.masterSM.checkMasterDropped(member.id().getId());
			}
		}

		@Override
		public void onReceiveMessage(Message message) {
			if (this.eventLoop.inEventLoop()) {
				this.instance.onReceiveMessage(message);
			} else {
				this.eventLoop.execute(() -> {
					this.onReceiveMessage(message);
				});
			}
		}
	}
}